package com.ruoyi.job.kettle;

import java.util.Random;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.Database;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleDatabaseException;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.logging.LogChannel;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.core.logging.StepLogTable;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.core.variables.VariableSpace;
import org.pentaho.di.core.variables.Variables;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.dummytrans.DummyTransMeta;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.sql.ExecSQLMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;


public class kettleUtil {

//	public void initKettleEnvironment(HttpServletRequest request) throws KettleException {
//		if (KettleEnvironment.isInitialized()) {
//			return;
//		}
//		/**
//		 * 为避免在部分网络环境中无法完成初始化，需要自行处理
//		 */
//		if (request == null) {
//			// 运行环境初始化
//			KettleEnvironment.init();
//		} else {
//			String userDir = System.getProperty("user.dir");
//			String kettleHome = request.getSession().getServletContext().getRealPath(File.separator   "WEB-INF");
//			// 设置用户路径和系统环境，包括用户路径和主目录
//			System.setProperty("user.dir", kettleHome);
//			System.setProperty("KETTLE_HOME", kettleHome);
//			// 运行环境初始化
//			KettleEnvironment.init();
//			// 避免造成影响其他程序的运行，还原用户路径
//			System.setProperty("user.dir", userDir);
//		}
//	}
	
	//添加配置数组，配置转化元
	public TransMeta buildTransMeta(String metaName, String... transXML) throws KettleXMLException {
		TransMeta transMeta = new TransMeta();
		// 设置转化元的名称
		transMeta.setName(metaName);
		// 添加转换的数据库连接
		for (int i = 0; i < transXML.length; i++ ) {
			transMeta.addDatabase(new DatabaseMeta(transXML[i]));
		}
		return transMeta;
	}
	
	//添加日志（可选操作）
	public void setStepLogTable(TransMeta transMeta, String connDbName, String tableName) {
		VariableSpace space = new Variables();
		// 将step日志数据库配置名加入到变量集中
		space.setVariable(Const.KETTLE_TRANS_LOG_DB, connDbName);
		space.initializeVariablesFrom(null);
		StepLogTable stepLogTable = StepLogTable.getDefault(space, transMeta);
		// 配置StepLogTable使用的数据库配置名称
		stepLogTable.setConnectionName(connDbName);
		// 设置Step日志的表名
		stepLogTable.setTableName(tableName);
		// 设置TransMeta的StepLogTable
		transMeta.setStepLogTable(stepLogTable);
	}
	
	//创建插件注册器
	public PluginRegistry getRegistry() {
	    // 插件注册，用于注册转换中需要用到的插件
	    return PluginRegistry.getInstance();
	}
	
	
	/**
	 * 该步骤用于获取源数据
	 * 设置表输入步骤
	 * @param transMeta
	 * @param registry
	 * @param sourceDbName
	 * @param sql
	 * @param stepName
	 * @return
	 */
	public StepMeta setTableInputStep(TransMeta transMeta, PluginRegistry registry, String sourceDbName, String sql,
			String stepName) {
		// 创建表输入
		TableInputMeta tableInputMeta = new TableInputMeta();
		String pluginId = registry.getPluginId(StepPluginType.class, tableInputMeta);
		// 指定数据源数据库配置名
		DatabaseMeta source = transMeta.findDatabase(sourceDbName);
		tableInputMeta.setDatabaseMeta(source);
		tableInputMeta.setSQL(sql);
		// 将表输入添加到转换中
		StepMeta stepMeta = new StepMeta(pluginId, stepName, tableInputMeta);
		// 给步骤添加在spoon工具中的显示位置
		stepMeta.setDraw(true);
		stepMeta.setLocation(100, 100);
		// 将表输入添加到步骤中
		transMeta.addStep(stepMeta);
		return stepMeta;
	}
	
	/**
	 * 设置表输出步骤，用于整表抽取
	 * @param transMeta
	 * @param registry
	 * @param targetDbName
	 * @param targetTableName
	 * @param stepName
	 * @return
	 */
	public StepMeta setTableOutput(TransMeta transMeta, PluginRegistry registry, String targetDbName,
			String targetTableName, String stepName) {
		// 创建表输出
		TableOutputMeta tableOutputMeta = new TableOutputMeta();
		String pluginId = registry.getPluginId(StepPluginType.class, tableOutputMeta);
		// 配置表输出的目标数据库配置名
		DatabaseMeta targetDb = transMeta.findDatabase(targetDbName);
		tableOutputMeta.setDatabaseMeta(targetDb);
		tableOutputMeta.setTableName(targetTableName);
		// 将表输出添加到转换中
		StepMeta stepMeta = new StepMeta(pluginId, stepName, tableOutputMeta);
		transMeta.addStep(stepMeta);
		return stepMeta;
	}
	 
	/**
	 * 更新步骤元
	 * 设置表插入与更新步骤，用于表中部分字段更新
	 * @param transMeta
	 * @param registry
	 * @param targetDbName
	 * @param targetTableName
	 * @param updatelookup lookup检索字段
	 * @param updateStream lookup更新字段
	 * @param updateStream2 lookup更新字段2
	 * @param conditions lookup条件
	 * @param updateOrNot lookup更新标记
	 * @param stepName
	 * @return
	 */
	public StepMeta setInsertUpdateMeta(TransMeta transMeta, PluginRegistry registry, String targetDbName,
			String targetTableName, String[] updatelookup, String[] updateStream, String[] updateStream2,
			String[] conditions, Boolean[] updateOrNot, String stepName) {
		// 创建插入与更新
		InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
		String pluginId = registry.getPluginId(StepPluginType.class, insertUpdateMeta);
		// 配置目标数据库配置名
		DatabaseMeta database_target = transMeta.findDatabase(targetDbName);
		insertUpdateMeta.setDatabaseMeta(database_target);
		// 设置目标表名
		insertUpdateMeta.setTableName(targetTableName);
		// 设置用来查询的关键字
		insertUpdateMeta.setKeyLookup(updatelookup);
		insertUpdateMeta.setKeyStream(updateStream);
		insertUpdateMeta.setKeyStream2(updateStream2);// 这一步不能省略
		insertUpdateMeta.setKeyCondition(conditions);
		// 设置要更新的字段
		insertUpdateMeta.setUpdateLookup(updatelookup);
		insertUpdateMeta.setUpdateStream(updateStream);
		insertUpdateMeta.setUpdate(updateOrNot);
		// 添加步骤到转换中
		StepMeta stepMeta = new StepMeta(pluginId, stepName, insertUpdateMeta);
		stepMeta.setDraw(true);
		stepMeta.setLocation(250, 100);
		transMeta.addStep(stepMeta);
		return stepMeta;
	}
	
	/**
	 * 绑定关联步骤
	 * 用于将表输入步骤与第二步骤绑定
	 * @param transMeta
	 * @param from
	 * @param to
	 */
	public void addTransHop(TransMeta transMeta, StepMeta from, StepMeta to) {
		transMeta.addTransHop(new TransHopMeta(from, to));
	}
	
	/**
	 * 执行数据抽取
	 * 执行抽取
	 * @param transMeta
	 * @param targetDbName
	 */
	public void executeTrans(TransMeta transMeta, String targetDbName) {
		try {
			Database database = new Database(null, transMeta.findDatabase(targetDbName));
			database.connect();
			Trans trans = new Trans(transMeta);
			trans.execute(new String[] { "start..." });
			trans.waitUntilFinished();
			// 关闭数据库连接
			database.disconnect();
			if (trans.getErrors() > 0) {
				throw new RuntimeException("There were errors during transformation execution.");
			}
		} catch (KettleDatabaseException e) {
			e.printStackTrace();
		} catch (KettleException e) {
			e.printStackTrace();
		}
	}
	
//	public static void main(String[] args) {
//		try {
//			KettleClient client = new KettleClient();
//			client.initEnvironment(null);
//			String transXML = ""; //  此处为上例的数据库配置
//			TransMeta meta = client.buildTransMeta("kettle", transXML);
//			PluginRegistry registry = client.getRegistry();
//			StepMeta step1 = client.setTableInputStep(meta, registry, "kettle", "select * from test1", "table input");
//			StepMeta step2 = client.setTableOutput(meta, registry, "kettle", "test2", "table insert");
//			client.addTransHop(meta, step1, step2);
//			client.executeTrans(meta, "kettle");
//		} catch (KettleException e) {
//			e.printStackTrace();
//		}
//	}
	
	public static void main(String[] args) throws KettleException {
		 
		KettleEnvironment.init();
		
		TransMeta transMeta = new TransMeta();
		transMeta.setName("QMCY_CUST_PERSONAL_INFO_IMPORT");
		
		//添加数据连接
		DatabaseMeta dataBaseInput = new DatabaseMeta("hive", "HIVE2", "Native", "32.172.33.2", "lsdb", "10000", "hdfs", "hdfs");
		DatabaseMeta dataBaseOutput = new DatabaseMeta("lsjzyx", "IBM DB2", "Native", "32.14.33.6", "lsjzyx", "60000", "db2inst1", "db2inst1");
		transMeta.addDatabase(dataBaseInput);
		transMeta.addDatabase(dataBaseOutput);
		
		//执行sql脚本
		ExecSQLMeta exSql = new ExecSQLMeta();
		exSql.setDatabaseMeta(transMeta.findDatabase("lsjzyx"));
		exSql.setSql("truncate table jzyx.QMCY_CUST_PERSONAL_INFO_IMPORT IMMEDIATE;");

		StepMeta sqlStep = new StepMeta("清空表数据",exSql);
		sqlStep.setLocation(50,50);
		sqlStep.setDraw(true);
		transMeta.addStep(sqlStep);
		
		//表输入
		TableInputMeta tableInputMeta = new TableInputMeta();
		tableInputMeta.setDatabaseMeta(transMeta.findDatabase("hive"));
		tableInputMeta.setSQL("SELECT\r\n" + 
				"	 SJRQ,ID_CARD,\r\n" + 
				"     NAME,\r\n" + 
				"	 CKYE,CKSNRJ,CKNRJ,CKJRJ,CKYRJ,DQYE,DQSNRJ,DQNRJ,\r\n" + 
				"	DQJRJ,DQYRJ,DKYE,DKSNRJ,DKNRJ,DKJRJ,DKYRJ,\r\n" + 
				"	LCYE,LCNRJ,SFXYK,SFSYB,SFSJYH,SFWSYH,SFSBK,SFSJWX,\r\n" + 
				"	SFCFT,SFZFB,SFETC,SFPOS,SFDXQY,SFYXDK,WLZFCPS,SLEEP_STS\r\n" + 
				"	FROM QMCY_CUST_PERSONAL_INFO");
		
		StepMeta inputStep = new StepMeta("输入步骤",tableInputMeta);
		inputStep.setLocation(100,50);
		inputStep.setDraw(true);
		transMeta.addStep(inputStep);
		
		// 创建表输出
		TableOutputMeta tableOutputMeta = new TableOutputMeta();
		// 配置表输出的目标数据库配置名
		tableOutputMeta.setDatabaseMeta(transMeta.findDatabase("lsjzyx"));
		tableOutputMeta.setSchemaName("JZYX");
		tableOutputMeta.setTableName("QMCY_CUST_PERSONAL_INFO_IMPORT");
		
		// 将表输出添加到转换中
		StepMeta outputStep = new StepMeta("输出步骤", tableOutputMeta);
		transMeta.addStep(outputStep);
		
		//空步骤
		DummyTransMeta dummyMeta = new DummyTransMeta();
		
		StepMeta dummyStep = new StepMeta("空步骤",dummyMeta);
		dummyStep.setLocation(200,50);
		dummyStep.setDraw(true);
		
		transMeta.addStep(dummyStep);
		
		//设置步骤直接的关系
		TransHopMeta hop0 = new TransHopMeta(sqlStep, inputStep);
		transMeta.addTransHop(hop0);
		TransHopMeta hop = new TransHopMeta(inputStep, outputStep);
		transMeta.addTransHop(hop);
		TransHopMeta hop2 = new TransHopMeta(outputStep, dummyStep);
		transMeta.addTransHop(hop2);
		
		Random random = new Random();
        int i = random.nextInt(1000);
        
		//开始执行
		Trans trans = new Trans(transMeta);
		
		LogChannel logChannel = new LogChannel(trans.getName() + i + System.currentTimeMillis());
        trans.setLog(logChannel);
        
		trans.setLogLevel(LogLevel.BASIC);
		trans.prepareExecution(null);
	
		trans.startThreads();
		trans.waitUntilFinished();
		
		String logText = KettleLogUtil.getLogText(trans.getLogChannelId(), true, trans.getLogDate().getTime());
		
		System.out.println(logText);
		 
		if(trans.getErrors() > 0) {
			System.out.println(">>>>>>>>>> ERROR" + logText);
		}else {
			System.out.println(">>>>>>>>>> SUCCESS ");
		}
		
	}
}
